package com.siteseat.crudeq.broker.netty;

import com.siteseat.crudeq.broker.transport.BrokerSession;
import com.siteseat.crudeq.broker.transport.BrokerSessionManager;
import com.siteseat.crudeq.broker.transport.RequestHandlerRepository;
import com.siteseat.crudeq.common.raw.protocol.v1.ErrorCode;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.frame.FrameDecoder;

/**
 * Derived from netty IntegerHeaderFrameDecoder sample
 */
public class ExecutionHandler extends FrameDecoder {

  public static final Object       DONE             = new Object();

  public static final int          HEADER_BYTE_SIZE = 1/* request id(byte) */+ 4/* size(int) */;

  private BrokerSessionManager     sessionManager;

  private RequestHandlerRepository requestHandlerRepository;

  /**
   * @param sessionManager
   */
  public ExecutionHandler( BrokerSessionManager sessionManager, RequestHandlerRepository requestHandlerRepository ) {
    this.sessionManager = sessionManager;
    this.requestHandlerRepository = requestHandlerRepository;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected Object decode( ChannelHandlerContext ctx, Channel channel, ChannelBuffer input ) throws Exception {

    // Make sure if the header was received.
    if( input.readableBytes() < HEADER_BYTE_SIZE ) {
      // The header was not received yet - return null.
      // This method will be invoked again when more packets are received and
      // appended to the buffer.
      return null;
    }

    // The header is in the buffer.

    // Mark the current buffer position before reading the header because the
    // whole frame might not be in the buffer
    // yet. We will reset the buffer position to the marked position if there's
    // not enough bytes in the buffer.
    input.markReaderIndex();
    int start = input.readerIndex();

    // Read the header
    byte reqId = input.readByte();
    int bodySize = input.readInt();
    
    if( bodySize < 0 ) {
      // suspicious request
      channel.close();
      return DONE;
    }

    // Make sure if there's enough bytes in the buffer.
    if( input.readableBytes() < bodySize ) {
      // The whole bytes were not received yet - return null.
      // This method will be invoked again when more packets are received and
      // appended to the buffer.

      // Reset to the marked position to read the length field again next time.
      input.resetReaderIndex();
      return null;
    }

    try {
      executeRequest( reqId, channel, input, bodySize );
      return DONE;
    } finally {
      // ensure we consume all these bytes despite if or not the execution succeeds.
      input.readerIndex( start + HEADER_BYTE_SIZE + bodySize );
    }
  }

  /**
   * @param channel
   * @param input
   * @param bodySize
   * @return
   * @throws Exception
   * @throws InvalidRequestException
   */
  protected void executeRequest( byte reqId, Channel ch, ChannelBuffer input, int bodySize ) throws Exception {
    BrokerSession s = this.sessionManager.get( ch.getId() );
    assert s != null;

    boolean authenticated = s.isAuthenticated();

    short ver;
    if( authenticated ) {
      ver = s.getVersion();
    } else {
      ver = input.readShort();
      if( ver <= 0 ) {
        CodecHelper.writeError( ch, ErrorCode.INVALID_REQUEST_VERSION );
        return;
      }
    }

    RequestHandler r = this.requestHandlerRepository.get( ver );
    if( r.isEnabled() == false ) {
      CodecHelper.writeError( ch, ErrorCode.SERVER_NOT_ENABLED_YET );
      return;
    }    
    if( ver != r.getVersion() ) {
      CodecHelper.writeError( ch, ErrorCode.INVALID_REQUEST_VERSION );
      return;
    }

    if( r.needAuthentication( s, ch, input, reqId ) ) {
      if( s.isAuthenticated() == false ) {
        CodecHelper.writeError( ch, ErrorCode.NOT_AUTHENTICATED_YET );
        return;
      }
    }

    r.execute( s, reqId, ch, input, ver );
  }

}
